package com.sansec.core.utils;

import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/** MQ工具类，整合mq资源，提供统一的消息发送和订阅入口
 * @author lichangwei
 */
@Component
public class RocketMqUtils {

    @Value(value = "${spring.cloud.stream.rocketmq.binder.name-server}")
    private String rocketMqAddress;

    /**
     * producer Map
     */
    private final static Map<String, MQProducer> MQPRODUCERMAP=new ConcurrentHashMap();

    public void send(Message message) throws NullPointerException, MQClientException, RemotingException, InterruptedException, MQBrokerException {
        if(message==null)
        {
            throw new NullPointerException("message is null");
        }
        MQProducer producer = getMQProducer("ccsp");
        producer.send(message);
    }

    /** 获取MQProducer
     * @param producerGroup producerGroup名称
     * @return MQProducer
     */
    private MQProducer getMQProducer(String producerGroup) throws MQClientException {
        if(StringUtils.isEmpty(producerGroup)){
            throw new NullPointerException("producerGroup 为空");
        }
        if(MQPRODUCERMAP.containsKey(producerGroup)){
            return MQPRODUCERMAP.get(producerGroup);
        }
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr( rocketMqAddress );
        producer.start();
        MQPRODUCERMAP.put(producerGroup,producer);
        return producer;
    }
}
